package com.lagou;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class DelProcessor extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {

        HTableInterface relations = e.getEnvironment().getTable(TableName.valueOf("relations"));
        byte[] row = delete.getRow();
        Set<Map.Entry<byte[], List<Cell>>> entries = delete.getFamilyCellMap().entrySet();
        for (Map.Entry<byte[], List<Cell>> entry : entries) {
            System.out.println("------------" + Bytes.toString(entry.getKey()));
            List<Cell> values = entry.getValue();
            for (Cell value : values) {
                byte[] rowkey = CellUtil.cloneRow(value);
                byte[] column = CellUtil.cloneQualifier(value);
                System.out.println(" rowkey------------" + rowkey);
                System.out.println(" column------------" + column);
//                判断要删除的数据是否存在
                boolean flag = relations.exists(new Get(column).addColumn(Bytes.toBytes("friends"), rowkey));

                if (flag) {
                    System.out.println("begin delete---------" + column + "--" + rowkey);
                    Delete deleteF = new Delete(column).addColumn(Bytes.toBytes("friends"), rowkey);
                    relations.delete(deleteF);
                }
            }
        }
        relations.close();
    }

}
